package com.xl.flinkdemo.transform;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @ClassName Transformtest1_base
 * @Description TODO
 * @Author hxl
 * @Date 2021/4/14 16:29
 * Version 1.0
 **/
public class Transformtest1_base {

  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    DataStream<String> stringDataStream =
        env.readTextFile("D:\\studyspace\\flinkdemo\\src\\main\\resources\\sensor.txt");


    //map
    DataStream<Integer> mapStream = stringDataStream.map(new MapFunction<String, Integer>() {
      @Override
      public Integer map(String s) throws Exception {
        return s.length();
      }
    });


    DataStream<String> flatMapStream =
        stringDataStream.flatMap(new FlatMapFunction<String, String>() {
          @Override
          public void flatMap(String s, Collector<String> collector) throws Exception {
            //以逗号分割
            String[] fields = s.split(",");
            for (String filed:fields){
              collector.collect(filed);
            }
          }
        });

    DataStream<String> fliterMap = stringDataStream.filter(new FilterFunction<String>() {
      @Override
      public boolean filter(String s) throws Exception {
        //筛选出id为sensor_1的value
        return s.startsWith("sensor_1");
      }
    });

    mapStream.print("map");
    flatMapStream.print("flatmap");
    fliterMap.print("flitermap");

    env.execute();
  }
}
